package common

import (
	"context"
	"encoding/hex"
	"fmt"
	"sync/atomic"
	"time"

	"github.com/golang/protobuf/proto"
	"hundsun.com/hsl/hschain/common/crypto"
	"hundsun.com/hsl/hschain/common/log/logging"
	"hundsun.com/hsl/hschain/common/message"
	"hundsun.com/hsl/hschain/common/types"
	"hundsun.com/hsl/hschain/common/util/pb"
	consensustypes "hundsun.com/hsl/hschain/consensus/types"
	"hundsun.com/hsl/hschain/contract/native/reward"
	"hundsun.com/hsl/hschain/protos"
	protoscommon "hundsun.com/hsl/hschain/protos/common"
	pbledger "hundsun.com/hsl/hschain/protos/ledger"
	"hundsun.com/hsl/hschain/store/mq"
)

var genesisBlock *pbledger.Block

// InitChecker 初始化checker相关流程
func (bc *BaseClient) InitChecker(conf *consensustypes.CheckerConfig) error {
	bc.checkerConf = conf
	bc.checkedBlkQueue = make(chan *consensustypes.CheckedBlock, 2*conf.PendingBlockPoolFetchSize)
	bc.maxRecvBlkSeq = &atomic.Value{}
	bc.maxRecvBlkSeq.Store(uint64(0))
	bc.lastCheckedBlkSeq = &atomic.Value{}
	bc.lastCheckedBlkSeq.Store(uint64(0))
	genesisBlock = pb.GetGenesisBlock()

	err := bc.syncBlockCheckedMsgWithLedger()
	if err != nil {
		return err
	}
	return bc.subBlockReceivedMessage()
}

func (bc *BaseClient) subBlockReceivedMessage() error {
	mqLog := bc.log.WithField(logging.LogFieldMessageTopic, message.TopicReceivedBlock)
	mqLog.Info("fetching latest message")
	latestMsgBytes, err := bc.RequestTopicInfo(types.EventQueryReceivedBlockTopic)
	if err != nil {
		return err
	}

	if len(latestMsgBytes) > 0 {
		var latestMsg protoscommon.BlockCheckedMessage
		err = proto.Unmarshal(latestMsgBytes, &latestMsg)
		if err != nil {
			return err
		}
		bc.maxRecvBlkSeq.Store(latestMsg.BlockNumber)
		mqLog.Infof("got latest message and set maxRecvBlkSeq:%d", latestMsg.BlockNumber)
	} else {
		mqLog.Info("no latest message")
	}
	mqLog.Info("subscribing")

	bc.SubReceivedBlockFunc(func(block *pbledger.Block) error {
		bc.log.Infof("recv new block %d and current maxRecvBlkSeq:%d", block.Header.Number, bc.maxRecvBlkSeq.Load())
		if block.Header.Number > bc.maxRecvBlkSeq.Load().(uint64) {
			bc.maxRecvBlkSeq.Store(block.Header.Number)
		}
		return nil
	})

	return err
}

// Run 拉取并验证块
func (bc *BaseClient) Run(ctx context.Context, startedSignal chan<- struct{}) {
	blockFetchIntervalSeconds := bc.checkerConf.BlockFetchInterval / consensustypes.MilliSecondsPerSecond
	if blockFetchIntervalSeconds < 1 {
		blockFetchIntervalSeconds = 1
	}
	bc.log.Infof("running general checker loop...")
	defer bc.log.Info("exit  general checker loop...")
	lastHandleTime := time.Now().Unix()

	sweeperCtx, _ := context.WithCancel(ctx)
	sweeperLog := bc.log.WithField(logging.LogFieldGoroutine, "sweeper")
	go bc.sweepAfterBlockCheckedFromBlockChain(sweeperCtx, sweeperLog)

	startedSignal <- struct{}{}
	for {
		select {
		case <-time.After(time.Duration(bc.checkerConf.BlockCheckInterval) * time.Millisecond):
			now := time.Now().Unix()
			if bc.maxRecvBlkSeq.Load().(uint64) > bc.lastCheckedBlkSeq.Load().(uint64) {
				_ = bc.tryToFetchAndCheckBlocks(bc.lastCheckedBlkSeq.Load().(uint64)+1, bc.maxRecvBlkSeq.Load().(uint64)+1, true)
				lastHandleTime = time.Now().Unix()
			} else if now-lastHandleTime >= int64(blockFetchIntervalSeconds) {
				_ = bc.tryToFetchAndCheckBlocks(bc.lastCheckedBlkSeq.Load().(uint64)+1, bc.lastCheckedBlkSeq.Load().(uint64)+uint64(bc.checkerConf.PendingBlockPoolFetchSize)+1, false)
				lastHandleTime = time.Now().Unix()
			}
		case <-ctx.Done():
			return
		}
	}
}

// tryToFetchAndCheckBlocks 尝试拉取[startBlockNumber,endBlockNumber)之间的块并进行验证连续的块
func (bc *BaseClient) tryToFetchAndCheckBlocks(startBlockSeq, endBlockSeq uint64, forceFetch bool) error {
	start, end := startBlockSeq, endBlockSeq
	bc.log.Debugf("try to fetch blocks[%d,%d)", startBlockSeq, endBlockSeq)
	for {
		if end-start > uint64(bc.checkerConf.PendingBlockPoolFetchSize) {
			end = start + uint64(bc.checkerConf.PendingBlockPoolFetchSize)
		}
		bc.log.Debugf("fetching blocks seq=[%d,%d)", start, end)
		fetchedBlocks, err := bc.RequestPendingBlocks(start, end-1)
		bc.log.Debugf("got %d blocks", len(fetchedBlocks))
		if err != nil {
			bc.log.Debugf("request pending blocks failed, err: %s", err.Error())
			return err
		}
		for _, blk := range fetchedBlocks {
			blkLog := pb.GetBlockLog(bc.log, blk)
			err = bc.CheckBlock(blk)
			if err != nil {
				err = removePendingBlock(blk.Header.Number, bc.GetTopicClient())
				if err != nil {
					bc.log.Errorf("Remove pending block failed, err: %s", err.Error())
				}
			}
			err = bc.SaveBlock(blk.Header.Number)
			if err != nil {
				blkLog.Errorf("save block failed, err: %s", err.Error())
			}
			bc.lastAcceptedBlock = blk
			if bc.checkerConf.BlockAcceptedCallback != nil {
				bc.checkerConf.BlockAcceptedCallback(blk)
			}
			bc.lastCheckedBlkSeq.Store(blk.Header.Number)
			bc.checkedBlkQueue <- &consensustypes.CheckedBlock{
				Block:  blk,
				Passed: err == nil,
			}
			blkLog.Infof("put into checkedBlkQueue")
			blkChkMsg := pb.FormatBlockCheckedMsg(bc.lastAcceptedBlock)
			blkChkMsgBytes, _ := proto.Marshal(blkChkMsg)

			if blkLog.IsTraceEnabled() {
				for _, txHash := range pb.GetTxHashesInBlock(blk) {
					blkLog.WithField(logging.LogFieldTxHash, txHash.GetShortString()).Tracef("CheckBlock passed:%v", err == nil)
				}
			}

			mqLog := blkLog.WithField(logging.LogFieldMessageTopic, message.TopicCheckedBlock)
			err = bc.StoreTopicInfo(blkChkMsgBytes, types.EventStoreCheckedBlockTopic)
			if err != nil {
				mqLog.Error("published failed")
			}
			mqLog.Infof("block %d published", blkChkMsg.BlockNumber)
			bc.UpdateLastBlockCheckedMessage(blk)
		}

		// 如果处理完了则结束
		if end >= endBlockSeq {
			return nil
		}
		start = end
		end = endBlockSeq
	}

}

// CheckBlock 验证区块
func (bc *BaseClient) CheckBlock(blk *pbledger.Block) (err error) {
	blkLog := pb.GetBlockLog(bc.log, blk)
	blkLog.Debug("checker started...")
	defer func() {
		if err == nil {
			bc.log.Debug("accepted")
		} else {
			bc.log.Errorf("rejected: %s", err.Error())
		}

	}()
	err = bc.preValidateBlock(blk, bc.lastAcceptedBlock)
	if err != nil {
		return err
	}
	if bc.checkerConf.BlockCreatorValidator != nil {
		err = bc.checkerConf.BlockCreatorValidator(blk)
		if err != nil {
			return err
		}
	}

	return bc.validateCheckBlockExecResult(blk)
}

func (bc *BaseClient) validateCheckBlockExecResult(blk *pbledger.Block) error {
	checkBlockNumber, checkBlockEnabled := pb.GetCheckBlockNumber(blk.Header.Number)
	if checkBlockEnabled {
		t4 := time.Now()
		var checkBlockExecMsg *protoscommon.BlockExecutedMessage
		for checkBlockExecMsg == nil {
			checkBlockExecMsg = bc.GetBlockExecutedMessage(checkBlockNumber, bc.log)
		}
		bc.log.Debugf("get exec result of CheckBlock block,cost:%dms", time.Since(t4)/time.Millisecond)
		stateHash := protos.Hash(checkBlockExecMsg.StateHash)
		if !stateHash.Equals(blk.Header.StateHashOfCheckBlock) {
			return fmt.Errorf("unexpected CheckBlock block hash,expected:%s,actual:%s", hex.EncodeToString(blk.Header.StateHashOfCheckBlock), hex.EncodeToString(checkBlockExecMsg.StateHash))
		}
	}
	return nil
}
func (bc *BaseClient) syncBlockCheckedMsgWithLedger() error {
	bc.log.Info("start sync BlockCheckedMessage with ledger")
	bc.log.Info("fetching latest BlockCheckedMessage")

	lastBlock, err := bc.RequestLastBlock()
	if err == types.ErrNotFound {
		bc.fireBlockCheckedMessage(genesisBlock)
		bc.lastAcceptedBlock = genesisBlock
		bc.lastCheckedBlkSeq.Store(genesisBlock.Header.Number)
		return nil
	} else if err != nil {
		panic(fmt.Sprintf("request last block failed, err: %s", err.Error()))
	}

	bc.lastAcceptedBlock = lastBlock
	bc.lastCheckedBlkSeq.Store(lastBlock.Header.Number)

	var blockCheckedMsg protoscommon.BlockCheckedMessage
	blockCheckedMsgBytes, err := bc.RequestTopicInfo(types.EventQueryCheckedBlockTopic)
	if err != nil {
		panic(fmt.Sprintf("request topic failed :%s", err))
	}

	err = proto.Unmarshal(blockCheckedMsgBytes, &blockCheckedMsg)
	if err != nil {
		return err
	}

	if lastBlock.Header.Number > blockCheckedMsg.GetBlockNumber() {
		bc.fireBlockCheckedMessage(lastBlock)
	}
	bc.log.Infof("latest BlockCheckedMessage[seq=%d,number=%d]", blockCheckedMsg.BlockSequence, blockCheckedMsg.BlockNumber)
	return nil
}

func (bc *BaseClient) fireBlockCheckedMessage(block *pbledger.Block) error {
	msg := pb.FormatBlockCheckedMsg(block)
	msgBytes, err := proto.Marshal(msg)
	if err != nil {
		return err
	}
	return bc.StoreTopicInfo(msgBytes, types.EventStoreCheckedBlockTopic)
}

// preValidateBlock 预校验：除签名外必须字段格式、上一块有关信息（区块号连续、block.BlockTime>lastBlkTime、parentHash）
func (bc *BaseClient) preValidateBlock(currentBlk, lastBlk *pbledger.Block) error {
	err := bc.ValidateBasicInfoOfBlock(currentBlk)
	if err != nil {
		return err
	}
	err = pb.ValidatePrevBlockInfoOfBlock(currentBlk, lastBlk.Header.BlockTime, lastBlk.Header.Number, lastBlk.Hash())
	if err != nil {
		return err
	}
	return nil
}

func (bc *BaseClient) sweepAfterBlockCheckedFromBlockChain(ctx context.Context, log logging.Log) {
	log.Info("start sweep after block checked...")
	defer log.Info("exit sweep after block checked...")
	for {
		select {
		case item := <-bc.checkedBlkQueue:
			blkLog := pb.GetBlockLog(log, item.Block)
			txHashes := pb.GetTxHashesInBlock(item.Block)
			if item.Passed {
				blkLog.Infof("RemoveTransactions start..")
				if blkLog.IsTraceEnabled() {
					for _, txHash := range txHashes {
						blkLog.WithField(logging.LogFieldTxHash, txHash.GetShortString()).Trace("will be removed")
					}
				}

				err := bc.removePendingTransactions(txHashes)
				if err != nil {
					blkLog.Errorf("RemoveTransactions failed: %s", err.Error())
				} else {
					blkLog.Infof("RemoveTransactions ends")
				}
			} else {
				blkLog.Infof("UndoPreRemoveTransactions[CheckBlock failed]  starts...")
				_, err := bc.undoPreRemoveTransactions(txHashes)
				if err != nil {
					blkLog.Errorf("UndoPreRemoveTransactions[CheckBlock failed] failed: %s", err.Error())
				} else {
					blkLog.Infof("UndoPreRemoveTransactions[CheckBlock failed]  ends...")
					if blkLog.IsTraceEnabled() {
						for i := range txHashes {
							blkLog.WithField(logging.LogFieldTxHash, txHashes[i].GetShortString()).Trace("UndoPreRemoveTransaction")
						}
					}
				}
			}
		case <-ctx.Done():
			return
		}
	}
}

func (bc *BaseClient) removePendingTransactions(delHashes []protos.Hash) error {
	msg := bc.GetWorkerClient().NewMessage(types.P2pModule, mq.ModeWorker.String(), types.EventTxPoolRemoveTxs, delHashes)
	err := bc.GetWorkerClient().Send(msg, true)
	if err != nil {
		bc.log.Errorf("Client send new message failed, err: %s\n", err.Error())
		return err
	}
	_, err = bc.GetWorkerClient().WaitTimeout(msg, time.Second*consensustypes.DefaultMqTimeOut)
	if err != nil {
		bc.log.Errorf("RemovePendingTransactions failed, err: %s\n", err.Error())
		return err
	}

	return nil
}

func (bc *BaseClient) undoPreRemoveTransactions(undoDelHashes []protos.Hash) ([]int64, error) {
	msg := bc.GetWorkerClient().NewMessage(types.P2pModule, mq.ModeTopic.String(), types.EventTxPoolUndoPreRemove, undoDelHashes)
	err := bc.GetWorkerClient().Send(msg, true)
	if err != nil {
		return nil, err
	}
	resp, err := bc.GetWorkerClient().WaitTimeout(msg, time.Second*consensustypes.DefaultMqTimeOut)
	if err != nil {
		return nil, err
	}
	res := resp.Data.([]int64)
	return res, nil
}

func removePendingBlock(blockNum uint64, client mq.Client) error {
	msg := client.NewMessage(types.BlockChainModule, mq.ModeTopic.String(), types.EventDelPendingPoolBlock, blockNum)
	err := client.Send(msg, false)
	if err != nil {
		return err
	}

	return nil
}

// ValidateBasicInfoOfBlock 验证区块基本信息（未背书和已背书的块通用），验证内容如下：
// 1. 区块头各字段格式（不含共识字段和序列号字段）
// 2. 块中必须有交易
// 注意：此方法不适用创世块的校验！
func (bc *BaseClient) ValidateBasicInfoOfBlock(block *pbledger.Block) error {
	//触发初始化
	_ = pb.GetGenesisBlock()

	if block.Header == nil {
		return types.ErrNoBlockHeader
	}

	if block.Header.Number == 0 {
		return types.ErrNonPositiveBlockNumber
	}

	hashSize, err := crypto.HashSize()
	if err != nil {
		return err
	}
	if len(block.Header.ParentHash) != hashSize {
		return types.ErrInvalidParentHashSize
	}

	if block.Header.BlockTime < genesisBlock.Header.BlockTime {
		return types.ErrBlockTimeBeforeGenesisBlock
	}

	// 判断区块是否包含挖矿交易
	if bc.GetCurParam().GetFeeLimit().GetMinTxFee() != 0 {
		if len(block.Transactions) == 0 {
			return types.ErrEmpty
		}
		miningTx := block.Transactions[0]
		if miningTx.Ty != reward.NativeReward {
			return types.ErrInvalidTxType
		}
		if miningTx.GetValue() != bc.GetCurParam().GetReward() {
			return types.ErrInvalidParam
		}
	}

	checkBlockNumber, checkBlockEnabled := pb.GetCheckBlockNumber(block.Header.Number)
	if checkBlockEnabled && checkBlockNumber != block.Header.CheckBlock {
		return types.ErrInvalidCheckBlockNumber
	}
	if checkBlockEnabled && len(block.Header.StateHashOfCheckBlock) != hashSize {
		return types.ErrInvalidCheckBlockStateHashSize
	}

	if block.Header.Signature == nil || len(block.Header.Signature.Signature) == 0 || len(block.Header.Signature.PublicKey) == 0 {
		return types.ErrInvalidBlockSignatureFormat
	}

	if len(block.Transactions) == 0 {
		return types.ErrNoTransactions
	}

	return nil
}
